
namespace Stee.CAP8.MessageBus.Kafka.Admin
{
    using System;
    using System.Collections.Generic;
    using System.Configuration;
    using System.Linq;
    using System.Threading.Tasks;
    using Confluent.Kafka;
    using Confluent.Kafka.Admin;
    using NLog;
    using Stee.CAP8.MessageBus.Interfaces.Delegates;
    using Stee.CAP8.MessageBus.Interfaces.Interfaces;
    using Stee.CAP8.MessageBus.Interfaces.Models;
    using Stee.CAP8.MessageBus.Kafka.Configurations;
    using Stee.CAP8.MessageBus.Kafka.Constants;
    using Stee.CAP8.MessageBus.Kafka.Utility;

    public sealed class AdminClientImpl : IAdmin
    {
        #region Static Variables

        private readonly static Logger Logger = LogManager.GetCurrentClassLogger();

        #endregion Static Variables


        #region Fields

        /// <summary>
        /// Admin client configuration
        /// </summary>
        private AdminClientConfig adminClientConfig;

        /// <summary>
        /// Kafka admin client
        /// </summary>
        private IAdminClient adminClient;

        /// <summary>
        /// Gets the admin client error handler
        /// </summary>
        private HandleBrokerError adminErrorHandler;

        /// <summary>
        /// Mutex lock
        /// </summary>
        private readonly object mutex = new object();

        #endregion Fields


        #region Properties

        /// <summary>
        /// Gets an optional identifier of a Kafka consumer(in a consumer group) that is passed to a Kafka broker with every request.
        /// The sole purpose of this is to be able to track the source of requests beyond just ip and port
        /// by allowing a logical application name to be included in Kafka logs and monitoring aggregates.
        /// </summary>
        public string ClientID { get; private set; } = string.Empty;

        /// <summary>
        /// Gets the flag indicating that admin client is initialized and ready
        /// </summary>
        public bool IsInitialized { get; private set; } = false;

        /// <summary>
        /// Gets the flag indicating if admin client is disposed
        /// </summary>
        public bool IsDisposed { get; private set; } = false;

        #endregion Properties


        #region Constructor

        /// <summary>
        /// Initializes a new instance of the AdminClientImpl class
        /// </summary>
        public AdminClientImpl() { }

        #endregion Constructor


        #region Interface Implementation

        /// <summary>
        /// Initializes the admin client
        /// </summary>
        /// <param name="brokerConfigurationJSON">JSON configuration file path</param>
        /// <param name="brokerErrorHandler">Error handler</param>
        /// <returns>Return TRUE if initialized else FALSE</returns>
        public bool Initialize(string brokerConfigurationJSON, HandleBrokerError brokerErrorHandler)
        {
            try
            {
                if (this.IsInitialized)
                {
                    throw new InvalidOperationException($"Admin client is already initialized. IsInit: {this.IsInitialized}");
                }

                this.adminErrorHandler = brokerErrorHandler ??
                    throw new ArgumentNullException($"Error handler for broker is null: {nameof(brokerErrorHandler)}");

                var adminConfigs = ConfigurationHelper.GetBrokerConfiguration(brokerConfigurationJSON, ConfigurationConstants.BrokerConfigAdminSection);
                this.InitAdminClient(adminConfigs);

                return this.IsInitialized;
            }
            catch (Exception ex)
            {
                Logger.Error(ex, ex.Message);
                throw;
            }
        }

        /// <summary>
        /// Initializes the admin client
        /// </summary>
        /// <param name="adminConfigs">Admin configuration</param>
        /// <param name="brokerErrorHandler">Error handler</param>
        /// <returns>Return TRUE if initialized else FALSE</returns>
        public bool Initialize(Dictionary<string, string> adminConfigs, HandleBrokerError brokerErrorHandler)
        {
            try
            {
                if (this.IsInitialized)
                {
                    throw new InvalidOperationException($"Admin client is already initialized. IsInit: {this.IsInitialized}");
                }

                this.adminErrorHandler = brokerErrorHandler ??
                    throw new ArgumentNullException($"Error handler for broker is null: {nameof(brokerErrorHandler)}");

                this.InitAdminClient(adminConfigs);

                return this.IsInitialized;
            }
            catch (Exception ex)
            {
                Logger.Error(ex, ex.Message);
                throw;
            }
        }

        /// <summary>
        /// Create new topics in message broker by specified topic configuration file.
        /// An equivalent dead-letter queue topic will be created for each topic
        /// </summary>
        /// <param name="configs">Topic configurations</param>
        /// <param name="dlqConfigs">Optional: Dead-letter queue topic configuration. If NULL, default configuration is used.</param>
        /// <returns>Return TRUE if created successfully else false</returns>
        public Task<bool> CreateNewTopic(IDictionary<string, string> configs, IDictionary<string, string> dlqConfigs = null)
        {
            this.CheckAdminClientState();

            if (configs == null)
            {
                throw new ArgumentNullException($"Topic configurations cannot be null. Param: {configs}");
            }
            else if (configs.Count < 3)
            {
                throw new ArgumentException($"Topic configurations is missing mandatory configurations. Config count: {configs.Count}");
            }

            /* 
             * Note:
             * Sonar S4457: Because of the way async/ await methods are rewritten by the compiler, any exceptions thrown during the parameters check will happen only when the task is observed.That could happen far away from the source of the buggy code or never happen for fire - and - forget tasks.
             * Solution: move validation into a sync method outside of this async method
             */

            return this.CreateNewTopicAsync(configs, dlqConfigs);
        }

        /// <summary>
        /// Create new topics in message broker by specified topic configuration file.
        /// An equivalent dead-letter queue topic will be created for each topic
        /// </summary>
        /// <param name="topicName">Topic name</param>
        /// <param name="partitionCount">Number of partitions</param>
        /// <param name="repFactor">Topic replication factor</param>
        /// <param name="configs">Other topic configurations</param>
        /// <param name="dlqConfigs">Optional: Dead-letter queue topic configuration.
        /// If NULL, default DLQ configuration is used.</param>
        /// <returns>Return TRUE if created successfully else false</returns>
        public async Task<bool> CreateNewTopicAsync(string topicName, int partitionCount, short repFactor,
            IDictionary<string, string> configs, IDictionary<string, string> dlqConfigs = null)
        {
            try
            {
                this.CheckAdminClientState();
                var topicConfiguration = GetTopicConfiguration(configs, topicName, partitionCount, repFactor);

                return await this.CreateTopicAsync(topicConfiguration, dlqConfigs).ConfigureAwait(false);
            }
            catch (CreateTopicsException ex)
            {
                Logger.Error(ex, ex.Message);
                return false;
            }
            catch (Exception ex)
            {
                Logger.Error(ex, ex.Message);
                throw;
            }
        }

        /// <summary>
        /// BUG: Topic deletion only on Windows OS will crash broker, https://issues.apache.org/jira/browse/KAFKA-1194
        /// Delete topic from message broker
        /// </summary>
        /// <param name="topicName">Topic name for deletion</param>
        /// <returns>Return TRUE if deleted successfully else FALSE</returns>
        private async Task<bool> DeleteTopicAsync(string topicName)
        {
            try
            {
                this.CheckAdminClientState();

                var topicStr = ValidateHelper.CheckString(topicName);

                var topicOptions = new DeleteTopicsOptions
                {
                    OperationTimeout = TimeSpan.FromSeconds(ConfigurationConstants.AdminOperationTimeoutSeconds),
                    RequestTimeout = TimeSpan.FromSeconds(ConfigurationConstants.AdminQueryTimeoutSeconds),
                };

                //// Delete topic
                await this.adminClient.DeleteTopicsAsync(new string[] { topicStr }, topicOptions).ConfigureAwait(false);

                return true;
            }
            catch (DeleteTopicsException ex)
            {
                Logger.Error(ex, ex.Message);
                return false;
            }
            catch (Exception ex)
            {
                Logger.Error(ex, ex.Message);
                throw;
            }
        }

        /// <summary>
        /// https://github.com/confluentinc/confluent-kafka-dotnet/issues/926, Describing group is not supported yet.
        /// Gets the configuration for specified consumer group configuration name
        /// </summary>
        /// <param name="topicName">Subscription topic</param>
        /// <param name="consumerGroup">Consumer group name</param>
        /// <returns>Return a mapping of settings specific to consumer group configuration</returns>
        private async Task<IDictionary<string, ConfigurationEntry>> DescribeConsumerGroupConfigAsync(TopicName topicName, string consumerGroup)
        {
            try
            {
                ValidateHelper.ValidateTopicName(topicName);

                var configRes = CreateConsumerGroupResource(consumerGroup);
                var describeOptions = new DescribeConfigsOptions
                {
                    RequestTimeout = TimeSpan.FromSeconds(ConfigurationConstants.AdminQueryTimeoutSeconds),
                };

                var describeResult = await this.DescribeConfigurationAsync(new ConfigResource[] { configRes }, describeOptions).ConfigureAwait(false);

                return describeResult?[configRes.Name];
            }
            catch (Exception ex)
            {
                Logger.Error(ex, ex.Message);
                throw;
            }
        }

        /// <summary>
        /// Gets the configuration for specified broker configuration name
        /// </summary>
        /// <param name="brokerID">Broker config name</param>
        /// <returns>Return a mapping of settings specific to broker configuration</returns>
        public async Task<IDictionary<string, ConfigurationEntry>> DescribeBrokerConfigAsync(string brokerID)
        {
            try
            {
                this.CheckAdminClientState();

                var configRes = CreateBrokerConfigResource(brokerID);
                var describeOptions = new DescribeConfigsOptions
                {
                    RequestTimeout = TimeSpan.FromSeconds(ConfigurationConstants.AdminQueryTimeoutSeconds),
                };

                var describeResult = await this.DescribeConfiguration(new ConfigResource[] { configRes }, describeOptions).ConfigureAwait(false);

                return describeResult?[configRes.Name];
            }
            catch (DescribeConfigsException ex)
            {
                Logger.Warn(ex, ex.Message);
                return null;
            }
            catch (KafkaException ex)
            {
                Logger.Error(ex, ex.Message);
                return null;
            }
            catch (Exception ex)
            {
                Logger.Error(ex, ex.Message);
                throw;
            }
        }

        /// <summary>
        /// Gets the configurations for specified topic
        /// </summary>
        /// <param name="topicName">Topic name</param>
        /// <returns>Return a mapping of settings specific to topic</returns>
        public async Task<IDictionary<string, ConfigurationEntry>> DescribeTopicAsync(string topicName)
        {
            try
            {
                this.CheckAdminClientState();

                var configRes = CreateTopicConfigResource(topicName);
                var describeOptions = new DescribeConfigsOptions
                {
                    RequestTimeout = TimeSpan.FromSeconds(ConfigurationConstants.AdminQueryTimeoutSeconds),
                };

                var describeResult = await this.DescribeConfiguration(new ConfigResource[] { configRes }, describeOptions).ConfigureAwait(false);

                return describeResult?[configRes.Name];
            }
            catch (DescribeConfigsException ex)
            {
                Logger.Warn(ex, ex.Message);
                return null;
            }
            catch (Exception ex)
            {
                Logger.Error(ex, ex.Message);
                throw;
            }
        }

        /// <summary>
        /// Implement IDisposable
        /// </summary>
        public void Dispose()
        {
            this.Dispose(true);
            GC.SuppressFinalize(this);
        }

        #endregion Interface Implementation


        #region Private Methods

        /// <summary>
        /// Initialize
        /// </summary>
        /// <param name="adminConfigs">Admin configuration</param>
        private void InitAdminClient(IDictionary<string, string> adminConfigs)
        {
            if (adminConfigs?.Any() != true)
            {
                throw new ConfigurationErrorsException(
                    $"No configuration found for admin client in section: {ConfigurationConstants.BrokerConfigAdminSection}");
            }
            else if (!adminConfigs.ContainsKey(ConfigurationConstants.AdminCfgBootstrapServers))
            {
                throw new ConfigurationErrorsException(
                    $"Bootstrap server configuration not found for admin client: {ConfigurationConstants.AdminCfgBootstrapServers}");
            }

            //// Create admin configuration
            var configuration = new AdminConfiguration(adminConfigs);

            this.adminClientConfig = new AdminClientConfig
            {
                BootstrapServers = configuration.BootstrapServers,
                //ClientId = configuration.ClientID,
            };

            this.InitializeAdminClient();
            this.IsInitialized = true;
        }

        /// <summary>
        /// Builds the admin client
        /// </summary>
        private void InitializeAdminClient()
        {
            var adminClientBuilder = new AdminClientBuilder(this.adminClientConfig);
            adminClientBuilder.SetErrorHandler(this.AdminClientErrorHandler);

            lock (this.mutex)
            {
                this.adminClient = adminClientBuilder.Build();
            }
        }

        /// <summary>
        /// Create new topics in message broker by specified topic configuration file.
        /// An equivalent dead-letter queue topic will be created for each topic
        /// </summary>
        /// <param name="configs">Topic configurations</param>
        /// <param name="dlqConfigs">Optional: Dead-letter queue topic configuration. If NULL, default configuration is used.</param>
        /// <returns>Return TRUE if created successfully else false</returns>
        private async Task<bool> CreateNewTopicAsync(IDictionary<string, string> configs, IDictionary<string, string> dlqConfigs = null)
        {
            try
            {
                //// Process mandatory configs and others
                var topicName = configs[TopicConfiguration.TopicConfigName];
                var partitionCount = configs[TopicConfiguration.TopicConfigNumOfPartitions];
                var repFactor = configs[TopicConfiguration.TopicConfigReplicationFactor];
                var topicConfiguration =
                    GetTopicConfiguration(configs, topicName, int.Parse(partitionCount), short.Parse(repFactor));

                return await this.CreateTopicAsync(topicConfiguration, dlqConfigs).ConfigureAwait(false);
            }
            catch (CreateTopicsException ex)
            {
                Logger.Error(ex, ex.Message);
                return false;
            }
            catch (Exception ex)
            {
                Logger.Error(ex, ex.Message);
                throw;
            }
        }

        /// <summary>
        /// Create topic asynchronously
        /// </summary>
        /// <param name="topicConfiguration">Topic configuration</param>
        /// <param name="dlqConfigs">DLQ topic configuration</param>
        /// <returns>Return TRUE if topic created successfully else FALSE</returns>
        private async Task<bool> CreateTopicAsync(TopicConfiguration topicConfiguration, IDictionary<string, string> dlqConfigs)
        {
            //// Create new topic specification
            var topicSpec = new TopicSpecification
            {
                Name = topicConfiguration.Name,
                NumPartitions = topicConfiguration.NumOfPartitions,
                ReplicationFactor = topicConfiguration.ReplicationFactor,
                Configs = topicConfiguration.GetAllConfigurations(),
            };

            var topicOptions = new CreateTopicsOptions
            {
                OperationTimeout = TimeSpan.FromSeconds(ConfigurationConstants.AdminOperationTimeoutSeconds),
                RequestTimeout = TimeSpan.FromSeconds(ConfigurationConstants.AdminQueryTimeoutSeconds),
                ValidateOnly = false,
            };

            //// Create new topic
            await this.adminClient.CreateTopicsAsync(
                new TopicSpecification[] { topicSpec }, topicOptions).ConfigureAwait(false);

            //// Note: Use partition count & replication factor of original topic for DLQ
            //// Create new DLQ topic
            topicSpec.Name = ConfigurationHelper.GenerateDLQTopicName(topicSpec.Name);
            topicSpec.Configs =
                (dlqConfigs ?? ConfigurationHelper.GetDLQTopicConfiguration()) as Dictionary<string, string>;

            await this.adminClient.CreateTopicsAsync(
                new TopicSpecification[] { topicSpec }, topicOptions).ConfigureAwait(false);

            return true;
        }

        /// <summary>
        /// Validate & describe configuration
        /// </summary>
        /// <param name="configResources">Configuration resources</param>
        /// <param name="options">Describe options</param>
        /// <returns>Return a mapping of settings specific to configuration resources</returns>
        private Task<IDictionary<string, IDictionary<string, ConfigurationEntry>>> DescribeConfiguration(
            IEnumerable<ConfigResource> configResources, DescribeConfigsOptions options)
        {
            if (configResources == null)
            {
                throw new ArgumentNullException(nameof(configResources));
            }
            else if (!configResources.Any())
            {
                return Task.FromResult<IDictionary<string, IDictionary<string, ConfigurationEntry>>>(null);
            }

            return this.DescribeConfigurationAsync(configResources, options);
        }

        /// <summary>
        /// Describe configuration asynchronously
        /// </summary>
        /// <param name="configResources">Configuration resources</param>
        /// <param name="options">Describe options</param>
        /// <returns>Return a mapping of settings specific to configuration resources</returns>
        private async Task<IDictionary<string, IDictionary<string, ConfigurationEntry>>> DescribeConfigurationAsync(
            IEnumerable<ConfigResource> configResources, DescribeConfigsOptions options)
        {
            var describeConfigResults =
                await this.adminClient.DescribeConfigsAsync(configResources, options).ConfigureAwait(false);

            return ProcessConfigDescriptions(describeConfigResults);
        }

        /// <summary>
        /// Validate topics in Kafka broker
        /// </summary>
        /// <param name="topicHash">Topic name hash</param>
        /// <returns>Return TRUE if topics are found else FALSE</returns>
        private async Task<bool> ValidateTopics(HashSet<string> topicHash)
        {
            var isValid = true;

            try
            {
                var topicResources = new List<ConfigResource>();
                var describeOptions = new DescribeConfigsOptions
                {
                    RequestTimeout = TimeSpan.FromSeconds(ConfigurationConstants.AdminQueryTimeoutSeconds),
                };

                foreach (var topic in topicHash)
                {
                    topicResources.Add(CreateTopicConfigResource(topic));
                }

                var describeResult = await this.DescribeConfiguration(topicResources, describeOptions).ConfigureAwait(false);

                if (describeResult?.Any() == true)
                {
                    //// Check for exact description responses 
                    foreach (var topic in topicHash)
                    {
                        isValid &= describeResult.ContainsKey(topic);
                    }
                }
                else
                {
                    isValid = false;
                }
            }
            catch (DescribeConfigsException ex)
            {
                Logger.Error(ex, ex.Message);
                isValid = false;
            }

            return isValid;
        }

        /// <summary>
        /// Set the handler to call on error events e.g. connection failures or all brokers down.
        /// Note that the client will try to automatically recover from errors that are not marked as fatal.
        /// Non-fatal errors should be interpreted as informational rather than catastrophic.
        /// </summary>
        /// <param name="iproducer">Producer instance</param>
        /// <param name="error">Kafka error</param>
        private void AdminClientErrorHandler(IAdminClient iClient, Error error)
        {
            this.adminErrorHandler?.Invoke(error.Code.ToString(), error.Reason, error.IsFatal);
        }

        /// <summary>
        /// Check current admin client state
        /// </summary>
        private void CheckAdminClientState()
        {
            if (!this.IsInitialized)
            {
                throw new InvalidOperationException($"Admin client is not initialized. IsInit: {this.IsInitialized}");
            }
            else if (this.IsDisposed || this.adminClient == null)
            {
                Logger.Error("Admin client have been disposed, create new instance required.");
                throw new TypeInitializationException(typeof(AdminClientImpl).FullName, null);
            }
        }

        /// <summary>
        /// Dispose
        /// </summary>
        /// <param name="disposing">If disposing equals true, the method has been called directly
        /// or indirectly by a user's code. Managed and unmanaged resources can be disposed.
        /// If disposing equals false, the method has been called by the runtime from inside the
        /// finalizer and you should not reference other objects. Only unmanaged resources can be disposed.</param>
        private void Dispose(bool disposing)
        {
            try
            {
                if (!this.IsDisposed && disposing)
                {
                    this.DisposeAdminClient();
                }

                this.IsDisposed = true;
                this.IsInitialized = false;
            }
            catch (Exception ex)
            {
                Logger.Info(ex, ex.Message);
                throw;
            }
        }

        /// <summary>
        /// Dispose admin client
        /// </summary>
        private void DisposeAdminClient()
        {
            lock (this.mutex)
            {
                if (this.adminClient != null)
                {
                    this.adminClient.Dispose();
                    this.adminClient = null;
                }
            }
        }

        #endregion Private Methods


        #region Static Methods

        /// <summary>
        /// Get topic configuration and validate topic names
        /// </summary>
        /// <param name="configs">Configuration collection</param>
        /// <param name="topicName">Topic name</param>
        /// <param name="partitionCount">Topic partition</param>
        /// <param name="repFactor">Topic replication factor</param>
        /// <returns>Return new topic configuration if valid</returns>
        private static TopicConfiguration GetTopicConfiguration(
            IDictionary<string, string> configs, string topicName, int partitionCount, short repFactor)
        {
            //// Create new topic configuration (internal validation)
            var topicConfiguration = new TopicConfiguration(topicName, partitionCount, repFactor, configs);

            //// Check if new topic name violates message bus naming convention (ends with "_dlq")
            if (ConfigurationHelper.CheckDLQTopicName(topicConfiguration.Name))
            {
                throw new ArgumentException(
                    $"New topic name cannot end with {ConfigurationConstants.TopicNamePostFixDLQ}, Topic name:{topicName}");
            }

            return topicConfiguration;
        }

        /// <summary>
        /// Process Kafka configuration description results
        /// </summary>
        /// <param name="describeConfigResults">Configuration description results</param>
        /// <returns>Return mapping of configuration collection per description</returns>
        private static IDictionary<string, IDictionary<string, ConfigurationEntry>> ProcessConfigDescriptions(List<DescribeConfigsResult> describeConfigResults)
        {
            if (describeConfigResults?.Any() == true)
            {
                var resultList = new Dictionary<string, IDictionary<string, ConfigurationEntry>>();

                foreach (var result in describeConfigResults)
                {
                    var topicSettings = new Dictionary<string, ConfigurationEntry>();

                    foreach (var item in result.Entries)
                    {
                        var entryResult = item.Value;
                        var configEntry = new ConfigurationEntry(entryResult.Name, entryResult.Value,
                            entryResult.IsDefault, entryResult.IsReadOnly, entryResult.IsSensitive);

                        topicSettings[item.Key] = configEntry;
                    }

                    resultList[result.ConfigResource.Name] = topicSettings;
                }

                return resultList;
            }
            else
            {
                return new Dictionary<string, IDictionary<string, ConfigurationEntry>>();
            }
        }

        /// <summary>
        /// Create config resource instance of consumer group
        /// </summary>
        /// <param name="consumerGroup">Consumer group name</param>
        /// <returns>Return consumer group config resource</returns>
        private static ConfigResource CreateConsumerGroupResource(string consumerGroup)
        {
            var configStr = ValidateHelper.CheckString(consumerGroup, excpMsg: "Consumer group name cannot be NULL or empty.");

            return new ConfigResource
            {
                Name = configStr,
                Type = ResourceType.Group,
            };
        }

        /// <summary>
        /// Create config resource instance of broker type
        /// </summary>
        /// <param name="brokerID">Broker config name</param>
        /// <returns>Return broker config resource</returns>
        private static ConfigResource CreateBrokerConfigResource(string brokerID)
        {
            var configStr = ValidateHelper.CheckString(brokerID, excpMsg: "Broker ID cannot be NULL or empty.");

            return new ConfigResource
            {
                Name = configStr,
                Type = ResourceType.Broker,
            };
        }

        /// <summary>
        /// Create config resource instance of topic type
        /// </summary>
        /// <param name="topicName">Topic name</param>
        /// <returns>Return topic config resource</returns>
        private static ConfigResource CreateTopicConfigResource(string topicName)
        {
            var topicStr = ValidateHelper.CheckString(topicName, excpMsg: "Topic name cannot be NULL or empty.");

            return new ConfigResource
            {
                Name = topicStr,
                Type = ResourceType.Topic,
            };
        }

        #endregion Static Methods
    }
}
